package com.shujia.stream.card

import org.apache.flink.table.api.{EnvironmentSettings, TableConfig, TableEnvironment}

object CardDayFlowIndex {
  def main(args: Array[String]): Unit = {
    //环境设置对象
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      //指定处理模式
      .inStreamingMode() //流处理模式
      //.inBatchMode()//批处理模式
      .build()

    //flink sql的执行环境
    val tEnv: TableEnvironment = TableEnvironment.create(settings)

    val config: TableConfig = tEnv.getConfig

    //增加优化参数
    config.set("table.exec.mini-batch.enabled", "true")
    config.set("table.exec.mini-batch.allow-latency", "5s")
    config.set("table.exec.mini-batch.size", "5000")
    config.set("table.optimizer.agg-phase-strategy", "TWO_PHASE")

    /**
     * 1、创建source source表
     *
     */
    tEnv.executeSql(
      """
        |CREATE TABLE cars (
        |    car STRING,
        |    city_code STRING,
        |    county_code STRING,
        |    card BIGINT,
        |    camera_id STRING,
        |    orientation STRING,
        |    road_id  BIGINT,
        |    `time` BIGINT,
        |    speed DOUBLE
        |) WITH (
        |  'connector' = 'kafka', -- 连接器类型，kafka
        |  'topic' = 'cars', -- kafka的topic
        |  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- kafk集群的列表
        |  'properties.group.id' = 'testGroup', -- 消费者组
        |  'scan.startup.mode' = 'earliest-offset', -- 读取数据位置
        |  'format' = 'json' -- 数据的格式，csv:文本格式。入股欧式用csv格式，字段的顺序必须和数据的顺序一致
        |)
        |""".stripMargin)

    /**
     * 2、创建sink 表
     */
    tEnv.executeSql(
      """
        |CREATE TABLE ads_card_day_flow_index (
        |    card BIGINT,-- 字段名和字段类型需要和mysql中表的字段名和类型保持一致
        |    tj_day STRING,
        |    flow BIGINT,
        |    PRIMARY KEY (card,tj_day) NOT ENFORCED
        |) WITH (
        |    'connector' = 'jdbc', -- 连接器类型
        |    'url' = 'jdbc:mysql://master:3306/car',-- jdbc地址
        |    'table-name' = 'ads_card_day_flow_index', -- 需要手动到mysql中创建表
        |    'username' = 'root',
        |    'password' = '123456'
        |)
        |
        |""".stripMargin)

    /**
     * 3、执行sql
     */
    tEnv.executeSql(
      """
        |insert into ads_card_day_flow_index
        |select
        |card,
        |date_format(FROM_UNIXTIME(`time`),'yyyy-MM-dd') as tj_day,
        |count(1) as flow
        |from
        |cars
        |group by card,date_format(FROM_UNIXTIME(`time`),'yyyy-MM-dd')
        |""".stripMargin)


  }
}
